<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="qWJgB" id="qWJgB"><span data-lake-id="ua13fbaba" id="ua13fbaba">典型回答</span></h1>
  <p data-lake-id="u7730ba80" id="u7730ba80"><span data-lake-id="uffe8774f" id="uffe8774f">当我们使用Kafka发送消息时，一般有两种方式，分别是同步发送（</span><code data-lake-id="u27819227" id="u27819227"><span data-lake-id="ue4187b26" id="ue4187b26">producer.send(msg).get()</span></code><span data-lake-id="uab39ac1b" id="uab39ac1b"> ）及异步发送（</span><code data-lake-id="u296902a5" id="u296902a5"><span data-lake-id="u2d50f76b" id="u2d50f76b">producer.send(msg, callback)</span></code><span data-lake-id="u91dc39ed" id="u91dc39ed">）。</span></p>
  <p data-lake-id="u64ccb9fb" id="u64ccb9fb"><br></p>
  <p data-lake-id="u34e76674" id="u34e76674"><span data-lake-id="uaa7468a5" id="uaa7468a5">同步发送的时候，可以在发送消息后，通过get方法等待消息结果：</span><code data-lake-id="uf053a706" id="uf053a706"><span data-lake-id="u6b45bdbd" id="u6b45bdbd">producer.send(record).get();</span></code><span data-lake-id="udaadd97e" id="udaadd97e"> ，这种情况能够准确的拿到消息最终的发送结果，要么是成功，要么是失败。</span></p>
  <p data-lake-id="ufcb1f69c" id="ufcb1f69c"><br></p>
  <p data-lake-id="udac02cab" id="udac02cab"><span data-lake-id="ucce77319" id="ucce77319">而异步发送，是采用了callback的方式进行回调的，可以大大的提升消息的吞吐量，也可以根据回调来判断消息是否发送成功。</span></p>
  <p data-lake-id="ud0da714f" id="ud0da714f"><br></p>
  <p data-lake-id="ua4fb880a" id="ua4fb880a"><span data-lake-id="uf1b55c43" id="uf1b55c43">不管是同步发送还是异步发送，最终都需要在Producer端把消息发送到Broker中，那么这个过程大致如下：</span></p>
  <p data-lake-id="u51de4f9d" id="u51de4f9d"><br></p>
  <p data-lake-id="ubbabb179" id="ubbabb179"><img src="https://cdn.nlark.com/yuque/0/2023/png/5378072/1684314627784-48da01a2-45c4-4a04-bb2e-1031018df8c7.png?x-oss-process=image%2Fwatermark%2Ctype_d3F5LW1pY3JvaGVp%2Csize_40%2Ctext_SmF2YSA4IEd1IFA%3D%2Ccolor_FFFFFF%2Cshadow_50%2Ct_80%2Cg_se%2Cx_10%2Cy_10"><img src="media/16843057638079/16843120535902.jpg?x-oss-process=image%2Fwatermark%2Ctype_d3F5LW1pY3JvaGVp%2Csize_9%2Ctext_SmF2YSA4IEd1IFA%3D%2Ccolor_FFFFFF%2Cshadow_50%2Ct_80%2Cg_se%2Cx_10%2Cy_10"></p>
  <p data-lake-id="u420adf89" id="u420adf89"><br></p>
  <p data-lake-id="u262aba31" id="u262aba31"><span data-lake-id="ub1313e93" id="ub1313e93">Kafka 的 Producer 在发送消息时通常涉及两个线程，</span><strong><span data-lake-id="ufcb8902c" id="ufcb8902c">主线程（Main）和发送线程（Sender）和一个消息累加器（RecordAccumulator）</span></strong></p>
  <p data-lake-id="u9303a9e2" id="u9303a9e2"><br></p>
  <p data-lake-id="u8cbfd29c" id="u8cbfd29c"><strong><span data-lake-id="ud1ed3887" id="ud1ed3887">Main线程</span></strong><span data-lake-id="u361211a7" id="u361211a7">是 Producer 的入口，负责初始化 Producer 的配置、创建 KafkaProducer 实例并执行发送逻辑。它会按照用户定义的发送方式（同步或异步）发送消息，然后等待消息发送完成。</span></p>
  <p data-lake-id="udc3e63be" id="udc3e63be"><br></p>
  <p data-lake-id="ua77d92e7" id="ua77d92e7"><span data-lake-id="u58ffec7e" id="u58ffec7e">一条消息的发送，在调用send方法后，会经过</span><strong><span data-lake-id="u8ea7fe77" id="u8ea7fe77">拦截器、序列化器及分区器</span></strong><span data-lake-id="ud8f317e6" id="ud8f317e6">。</span></p>
  <ul list="u774cd338">
   <li fid="u73f2a10a" data-lake-id="ua2c23c4c" id="ua2c23c4c"><span data-lake-id="u3f78d538" id="u3f78d538">拦截器主要用于在消息发送之前和之后对消息进行定制化的处理，如对消息进行修改、记录日志、统计信息等。</span></li>
   <li fid="u73f2a10a" data-lake-id="u5b67e1e1" id="u5b67e1e1"><span data-lake-id="u5b4a564f" id="u5b4a564f">序列化器负责将消息的键和值对象转换为字节数组，以便在网络上传输。</span></li>
   <li fid="u73f2a10a" data-lake-id="u98664ae3" id="u98664ae3"><span data-lake-id="u1281c428" id="u1281c428">分区器决定了一条消息被发送到哪个 Partition 中。它根据消息的键（如果有）或者特定的分区策略，选择出一个目标 Partition。</span></li>
  </ul>
  <p data-lake-id="u9d57cda1" id="u9d57cda1"><br></p>
  <p data-lake-id="uaa083b24" id="uaa083b24"><strong><span data-lake-id="ue3f7321d" id="ue3f7321d">RecordAccumulator</span></strong><span data-lake-id="u36de91c0" id="u36de91c0">在 Kafka Producer 中起到了消息积累和批量发送的作用，当 Producer 发送消息时，不会立即将每条消息发送到 Broker，而是将消息添加到 RecordAccumulator 维护的内部缓冲区中，RecordAccumulator 会根据配置的条件（如batch.size、linger.ms）对待发送的消息进行批量处理。</span></p>
  <p data-lake-id="ufc23bb01" id="ufc23bb01"><br></p>
  <p data-lake-id="uc298467a" id="uc298467a"><span data-lake-id="u881dd0a8" id="u881dd0a8">当满足指定条件时，RecordAccumulator 将缓冲区中的消息组织成一个批次（batch），然后一次性发送给 Broker。如果发送失败或发生错误，RecordAccumulator 可以将消息重新分配到新的批次中进行重试。这样可以确保消息不会丢失，同时提高消息的可靠性。</span></p>
  <p data-lake-id="u3cd4867f" id="u3cd4867f"><br></p>
  <p data-lake-id="u5589c95e" id="u5589c95e"><strong><span data-lake-id="ue561504e" id="ue561504e">Send线程</span></strong><span data-lake-id="ua78e066c" id="ua78e066c">是负责实际的消息发送和处理的。发送线程会定期从待发送队列中取出消息，并将其发送到对应的 Partition 的 Leader Broker 上。它主要负责网络通信操作，并处理发送请求的结果，包括确认的接收、错误处理等。</span></p>
  <p data-lake-id="ucd390e09" id="ucd390e09"><br></p>
  <p data-lake-id="u54a629af" id="u54a629af"><strong><span data-lake-id="u0daa0d07" id="u0daa0d07">NetworkClient 和 Selector </span></strong><span data-lake-id="ub28c4eca" id="ub28c4eca">是两个重要的组件，分别负责网络通信和 I/O 多路复用。</span></p>
  <p data-lake-id="u007ee92b" id="u007ee92b"><br></p>
  <p data-lake-id="uf038e8b8" id="uf038e8b8"><span data-lake-id="ub2d07598" id="ub2d07598">发送线程会把消息发送到Kafka集群中对应的Partition的Partition Leader，Partition Leader 接收到消息后，会对消息进行一系列的处理。它会将消息写入本地的日志文件（Log）</span></p>
  <p data-lake-id="ua8d7c4d9" id="ua8d7c4d9"><br></p>
  <p data-lake-id="u5d580aaa" id="u5d580aaa"><span data-lake-id="u4e1a3099" id="u4e1a3099">为了保证数据的可靠性和高可用性，Kafka 使用了消息复制机制。Leader Broker 接收到消息后，会将消息复制到其他副本（Partition Follower）。副本是通过网络复制数据的，它们会定期从 Leader Broker 同步消息。</span></p>
  <p data-lake-id="uc8ed733e" id="uc8ed733e"><br></p>
  <p data-lake-id="u3f7cf0a2" id="u3f7cf0a2"><span data-lake-id="uce846661" id="uce846661">每一个Partition Follower在写入本地log之后，会向Leader发送一个ACK。</span></p>
  <p data-lake-id="u1cd7901d" id="u1cd7901d"><br></p>
  <p data-lake-id="u55ab8e44" id="u55ab8e44"><span data-lake-id="u1eadbe36" id="u1eadbe36">但是我们的Producer其实也是需要依赖ACK才能知道消息有没有投递成功的，而这个ACK是何时发送的，Producer又要不要关心呢？这就涉及到了kafka的ack机制，生产者会根据设置的 request.required.acks 参数不同，选择等待或或直接发送下一条消息：</span></p>
  <p data-lake-id="u37019d41" id="u37019d41"><br></p>
  <ul list="u5c11ef18">
   <li fid="u7724da9b" data-lake-id="u26815f04" id="u26815f04"><span data-lake-id="u3c07b0bf" id="u3c07b0bf"> request.required.acks = 0 </span></li>
  </ul>
  <ul list="u5c11ef18" data-lake-indent="1">
   <li fid="u96265952" data-lake-id="u3b1695d5" id="u3b1695d5"><span data-lake-id="ud68e57ca" id="ud68e57ca">表示 Producer 不等待来自 Leader 的 ACK 确认，直接发送下一条消息。在这种情况下，如果 Leader 分片所在服务器发生宕机，那么这些已经发送的数据会丢失。</span></li>
  </ul>
  <ul list="u5c11ef18" start="2">
   <li fid="u7724da9b" data-lake-id="ufbee8899" id="ufbee8899"><span data-lake-id="uca37414a" id="uca37414a"> request.required.acks = 1 </span></li>
  </ul>
  <ul list="u5c11ef18" data-lake-indent="1">
   <li fid="ub037d996" data-lake-id="u5d6ab900" id="u5d6ab900"><span data-lake-id="uf475e475" id="uf475e475">表示 Producer 等待来自 Leader 的 ACK 确认，当收到确认后才发送下一条消息。在这种情况下，消息一定会被写入到 Leader 服务器，但并不保证 Follow 节点已经同步完成。所以如果在消息已经被写入 Leader 分片，但是还未同步到 Follower 节点，此时Leader 分片所在服务器宕机了，那么这条消息也就丢失了，无法被消费到。</span></li>
  </ul>
  <ul list="u5c11ef18" start="3">
   <li fid="u7724da9b" data-lake-id="uee9d3327" id="uee9d3327"><span data-lake-id="u1bf5d0f9" id="u1bf5d0f9"> request.required.acks = -1 </span></li>
  </ul>
  <ul list="u5c11ef18" data-lake-indent="1">
   <li fid="u2aa6c070" data-lake-id="ue054c60f" id="ue054c60f"><span data-lake-id="uf716180f" id="uf716180f">Leader会把消息复制到集群中的所有ISR（In-Sync Replicas，同步副本），要等待所有ISR的ACK确认后，再向Producer发送ACK消息，然后Producer再继续发下一条消息。</span></li>
  </ul>
 </body>
</html>